fastapi基础

入门

安装基础依赖

pip install fastapi

并且安装uvicorn来作为服务器：

pip install uvicorn[standard]

然后对你想使用的每个可选依赖项也执行相同的操作

启动服务

uvicorn main:app --reload

uvicorn main:app 命令含义如下:

• main：main.py 文件（一个 Python「模块」）。
• app：在 main.py 文件中通过 app = FastAPI() 创建的对象。
• --reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。

交互式api文档

http://127.0.0.1:8000/docs

备选api文档：http://127.0.0.1:8000/redoc

class Item(BaseModel):
    name: str
      # field中的example会覆盖docs上的示例，也会被config覆盖
    description: Optional[str] = Field(
        None, title="The description of the item", max_length=300, example="描述"
    )
    price: float = Field(..., gt=0, description="The price must be greater than zero")
    tax: Optional[float] = None

    class Config:
      # 会自动覆盖展示在docs上的示例
        schema_extra = {
                "example": {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                }
        }

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item = Body(
        ...,
        example={
            "name": "Foo",
            "description": "A very nice Item",
            "price": 35.4,
            "tax": 3.2,
        },
    ),
):
    results = {"item_id": item_id, "item": item}
    return results

fastapi的异步处理

from fastapi import FastAPI
import time
import asyncio
import os

app = FastAPI()

@app.get("/async_slowest")
async def async_slowest():
   time.sleep(1)
   return {"message": "async mode but use sync sleep"}

@app.get("/async_sleep_in_thread")
async def async_sleep_in_thread():
   loop = asyncio.get_event_loop()
   await loop.run_in_executor(None, time.sleep, 1)
   return {"message": "sleep run in thread pool"}

@app.get("/async_sleep")
async def async_sleep():
   await asyncio.sleep(1)
   return {"message": "async mode sleep"}

@app.get("/sync")
def sync_sleep():
   time.sleep(1)
   return {"message": "sync, but run in thread pool"}

我们用ab工具，总量100，并发100进行测试。

async

这4个函数，最慢的就是第一个async_slowest。
我们可以看到，它几乎是一个接一个的串联输出。
原因是：
fastapi框架会将async函数会放到event loop中运行。
如果函数没有运行或有await，则其他函数无法运行。
所以这里是一个串联的效果，总时间需要100s

async+loop.run_in_executor

为了解决这个问题，第二个函数引入了loop.run_in_executor
loop = asyncio.get_event_loop()
可以获取当前的event loop
loop.run_in_executor(None, time.sleep, 1)
是将time.sleep(1)放到一个线程池中去运行，所以不会出现阻塞。
这个函数，1秒可以全部运行完。

async+await

第三个函数是最正宗的实现。
它使用异步的sleep取代了原版同步的sleep。
这也是最快的实现。
1秒可以运行完

def

第四个函数是唯一一个不是async的普通函数。
它的运行时间是多少呢？
我的电脑是3秒运行完！
为什么？
这就是fastapi精彩的地方。
前面提到，async函数会放到event loop中执行。
那么，普通的函数会放到哪里呢？
答案是，放到thread pool中。
那么为什么是3秒呢。
这是因为我的电脑是逻辑8核。线程池的默认配置是核数*5，所以是40线程。
我的测试是100个并发，所以一共是3秒完成。
40->40->20

总结：简单的说，就像官方所说，如果你不清楚你函数里的调用是否异步，那就定义为普通函数。因为它可以采用多线程的方式解决。
反之，定义了async函数，里面却是同步的调用（第一个函数），那么这将慢的是灾难！

数据格式校验

Pydantic

额外的数据类型 https://fastapi.tiangolo.com/zh/tutorial/extra-data-types/ 日期，uuid，byte

路径

注意若前缀相同，可选参数应在默认路径之下，如下顺序不能互换

否则，/users/{user_id} 的路径还将与 /users/me 相匹配，”认为”自己正在接收一个值为 "me" 的 user_id 参数。

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}


@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

:path作为参数的类型表示可以接受任意参数，可用于404页面，用于文件路径

由于OpenApi不支持在路径参数中携带斜杠，因为这样会导致歧义，所以可以用Starlette 的一个内部工具在 FastAPI 中实现它

@app.get("/{path:path}")
def not_found():
    return 123

参数

可选参数及默认值

Optional表示这个参数是某个类型的可选参数，后面可以携带默认值

q: Optional[str] = None

布尔值参数

0,false,False,no 都是false

1,true,True,yes都是true

@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None,short:bool = False):
    return {"item_id": item_id, "q": q,"short":short}

多路径和参数

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: Optional[str] = None, short: bool = False
):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item

请求体+路径参数+查询参数

路径参数在路径后

请求体为一个json

查询参数为一个query

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Optional[str] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

多请求体参数

注意和单请求体区分，多请求体的中一个请求体的key会显示出来

class Item(BaseModel):
    name: str = None
    price: float = None
    is_offer: Optional[bool] = None


class User(BaseModel):
    username: str
    full_name: Optional[str] = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results

# 请求体参数
{
  "item_id": 123,
  "item": {
    "name": "string",
    "price": 0,
    "is_offer": true
  },
  "user": {
    "username": "hz",
    "full_name": null
  }
}

如果想要在多请求体参数后加单一参数

@app.put("/items/{item_id}")
async def update_item(
    item_id: int, item: Item, user: User, importance: int = Body(...)
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

# 请求体参数
{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "importance": 5
}

嵌入单个请求体参数

单个请求体参数是没有key的，如果我们非要给他加上key，Body提供一个可处理的参数

item: Item = Body(..., embed=True)

例子

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(..., embed=True)):
    results = {"item_id": item_id, "item": item}
    return results

#请求体参数
{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

嵌套参数

可以嵌套任意深度的模型

from typing import List, Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Image(BaseModel):
    url: HttpUrl
    name: str


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = set()
    images: Optional[List[Image]] = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results


# 请求体
{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "tags": [
        "rock",
        "metal",
        "bar"
    ],

    "images": [

        {
            "url": "http://example.com/baz.jpg",
            "name": "The Foo live"
        },
        {
            "url": "http://example.com/dave.jpg",
            "name": "The Baz"
        }
    ]
}

任意 dict 构成的请求体

from typing import Dict

from fastapi import FastAPI

app = FastAPI()


@app.post("/index-weights/")
async def create_index_weights(weights: Dict[int, float]):
    return weights

tips

请记住 JSON 仅支持将 str 作为键。

但是 Pydantic 具有自动转换数据的功能。

这意味着，即使你的 API 客户端只能将字符串作为键发送，只要这些字符串内容仅包含整数，Pydantic 就会对其进行转换并校验。

然后你接收的名为 weights 的 dict 实际上将具有 int 类型的键和 float 类型的值。

校验规则

实际上，Query、Path 和其他你将在之后看到的类，创建的是由一个共同的 Params 类派生的子类的对象，该共同类本身又是 Pydantic 的 FieldInfo 类的子类。

Pydantic 的 Field 也会返回一个 FieldInfo 的实例。

Body 也直接返回 FieldInfo 的一个子类的对象。还有其他一些你之后会看到的类是 Body 类的子类。

请记住当你从 fastapi 导入 Query、Path 等对象时，他们实际上是返回特殊类的函数。

Query

校验在？后的参数

通用的校验和元数据：

• alias 别名
• title
• description
• deprecated 是否显示弃用（默认false）

特定于字符串的校验：

• min_length
• max_length
• regex 正则匹配

完整实例用法

@app.get("/items")
async def create_item(q: Optional[str] = Query(
        None,
        title="参数q",
        description="参数q的描述",
        min_length=3,
        max_length=10,
        alias="items-q",
        regex="^f",
        deprecated=True
    )):
    return q

fastapi自带有参数格式校验规则，暂时只校验参数的格式，但是返回格式不是我们自定义的

{
    "detail": [
        {
            "loc": [
                "body",
                "price"
            ],
            "msg": "value is not a valid float",
            "type": "type_error.float"
        }
    ]
}

如果想要校验例如参数的最大长度等可以使用fastapi自带的Query

Query的第一个参数可用于设定默认值，当使用query设定默认值后，这个参数也会变成可选参数，所以optional实际上就没有用了

@app.get("/items")
async def create_item(q: Optional[str] = Query(None, min_length=3, max_length=50, regex="^fixedquery$")):
    return q

这个指定的正则表达式通过以下规则检查接收到的参数值：

• ^：以该符号之后的字符开头，符号之前没有字符。
• fixedquery: 值精确地等于 fixedquery。
• $: 到此结束，在 fixedquery 之后没有更多字符。

如果我们又要使用query又要让这个参数为必填参数可以使用三个点作为query的第一个参数

@app.get("/items")
async def create_item(q:str = Query(..., min_length=3, max_length=50, regex="^fixedquery$")):
    return q

{
    "detail": [
        {
            "loc": [
                "query",
                "q"
            ],
            "msg": "ensure this value has at most 10 characters",
            "type": "value_error.any_str.max_length",
            "ctx": {
                "limit_value": 10
            }
        }
    ]
}

Path

校验路径参数

对元数据扩展，和参数校验比query更多

• ge大于等于 gt大于 le 小于等于 lt 小于（适用于int和float）

Body

和path的校验参数相同，校验的是body体中的数据

Field

字段校验，校验规则和Body，Path相同

field不仅可以校验对象中的类型，也可以用于参数校验

注意，Field 是直接从 pydantic 导入的，而不是像其他的（Query，Path，Body 等）都从 fastapi 导入。

例子

class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None, title="The description of the item", max_length=300
    )
    price: float = Field(..., gt=0, description="The price must be greater than zero")
    tax: Optional[float] = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(..., embed=True)):
    results = {"item_id": item_id, "item": item}
    return results

Cookie

cookie的校验和上述校验规则都相同

@app.get("/items/")
async def read_items(ads_id: Optional[str] = Cookie(None)):
    return {"ads_id": ads_id}

Headers

@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}

请求头中会有一些参数是用减号-来连接的比如u-a，Python会自动转化这些参数

如果出于某些原因，你需要禁用下划线到连字符的自动转换，设置Header的参数 convert_underscores 为 False:

@app.get("/items/")
async def read_items(
    strange_header: Optional[str] = Header(None, convert_underscores=False)
):
    return {"strange_header": strange_header}

重复的请求头key

若存在多个x-token的请求头，在Python中可以通过list来接受

@app.get("/items/")
async def read_items(x_token: Optional[List[str]] = Header(None)):
    return {"X-Token values": x_token}

Form

用于表单校验

from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
async def login(*, username: str = Form(...), password: str = Form(...)):
     return {"username": username}

响应模型

请求类型装饰器参数

tips set不是必须，只是规范，使用其他类型也会被转成set，不影响

• response_model_include set 包含需要展示的字段

• response_model_exclude set 包含需要去除展示的字段

• response_model_exclude_unset=True 来仅返回显式设定的值

这里非常类似java中关于po，vo，dto等类似概念，传输数据和响应数据及时是同样的数据结构，也应该有不同的数据校验规则，所以应该有多种数据库映射类

比如用户传入用户名，密码，都是明文，我们返回给用户必须是密文，而数据库保存的数据字段可能会更多，如上三种不同的数据类型

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user

如果一张表中有很多字段，而我们只想返回有值的字段，可以使用参数response_model_exclude_unset=True

这样可以适用于某些场景，并不与上面这种响应模式冲突

from typing import List, Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: float = 10.5
    tags: List[str] = []


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}


@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

Union 或者 anyOf

你可以将一个响应声明为两种类型的 Union，这意味着该响应将是两种类型中的任何一种。

定义一个 Union 类型时，首先包括最详细的类型，然后是不太详细的类型。在下面的示例中，更详细的 PlaneItem 位于 Union[PlaneItem，CarItem] 中的 CarItem 之前。

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class BaseItem(BaseModel):
    description: str
    type: str


class CarItem(BaseItem):
    type = "car"


class PlaneItem(BaseItem):
    type = "plane"
    size: int


items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
}


@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
    return items[item_id]

常见的response对象，一般也就用JSONResponse

https://www.cnblogs.com/mazhiyong/p/13279543.html

合并预定response

from typing import Optional

from fastapi import FastAPI
from fastapi.responses import FileResponse
from pydantic import BaseModel


class Item(BaseModel):
    id: str
    value: str


responses = {
    404: {"description": "Item not found"},
    302: {"description": "The item was moved"},
    403: {"description": "Not enough privileges"},
}


app = FastAPI()


@app.get(
    "/items/{item_id}",
    response_model=Item,
    responses={**responses, 200: {"content": {"image/png": {}}}},
)
async def read_item(item_id: str, img: Optional[bool] = None):
    if img:
        return FileResponse("image.png", media_type="image/png")
    else:
        return {"id": "foo", "value": "there goes my hero"}

响应状态码

在请求装饰器中带status_code参数表示该次请求的响应状态码

status_code 也能够接收一个 IntEnum 类型，比如 Python 的 http.HTTPStatus

https://developer.mozilla.org/en-US/docs/Web/HTTP/Status

你可以使用来自 fastapi.status 的便捷变量。

内置的Json解析

from datetime import datetime

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel


class Item(BaseModel):
    title: str
    timestamp: datetime
    description: str = None


app = FastAPI()


@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
print(json_compatible_item_data)

点我就给你看点有意思的

• 本文作者：爱肖彤真是太好了
• 本文链接：http://huang_zhao.gitee.io/task/2021/07/02/python/%E6%A1%86%E6%9E%B6/fast-api/fastapi%20%E5%9F%BA%E7%A1%80/
• 版权声明：本博客所有文章除特别声明外，均默认采用 许可协议。